#include "destor.h"
#include "jcr.h"
#include "backup.h"
#include "utils/rabin_chunking.h"
#include "utils/threadpool.h"

static pthread_t resemblance_t;

static void *super_feature_thread(void *arg);

static void *resemblance_thread(void *arg)
{
    while (1)
    {
        // __attribute__((aligned(64))) unsigned long long int Feature[24];
        struct chunk *c = sync_queue_pop(dedup_queue);
        if (c == NULL)
            /* backup job finish */
            break;

        // sync_queue_push(resemblance_queue, c);
        // continue;

        if (CHECK_CHUNK(c, CHUNK_FILE_START) ||
            CHECK_CHUNK(c, CHUNK_FILE_END) ||
            CHECK_CHUNK(c, CHUNK_SEGMENT_START) ||
            CHECK_CHUNK(c, CHUNK_SEGMENT_END) ||
            CHECK_CHUNK(c, CHUNK_DUPLICATE))
        {
            sync_queue_push(resemblance_queue, c);
            continue;
            // free_chunk(c);
        }
        if (c->size > destor.chunk_min_size)
        {
            jcr.chunk_num_for_simi_dete++;
            jcr.total_size_for_similar_detection += c->size;
            TIMER_DECLARE(1);
            TIMER_BEGIN(1);
            if (destor.super_feature_method == SUPER_FEATURE)
                super_feature(c->data, c->size, c->superfeature);
            else if (destor.super_feature_method == SISD_FIXED_RABIN)
                Finesse(c->data, c->size, c->superfeature);
            else if (destor.super_feature_method == SISD_FIXED_GEAR)
                sisd_gear_fixed(c->data, c->size, c->superfeature);
            else if (destor.super_feature_method == SUPER_FEATURE_AVX)
                super_feature_avx(c->data, c->size, c->superfeature);
            else if (destor.super_feature_method == SUPER_FEATURE_AVX_FASTEST)
                super_feature_avx_fast(c->data, c->size, c->superfeature);
            else if (destor.super_feature_method == SISD_FIXED_RABIN_AVX)
                Finesse_avx(c->data, c->size, c->superfeature);
            else if (destor.super_feature_method == SISD_FIXED_RABIN_AVX_FASTEST)
                sisd_rabin_fixed_avx(c->data, c->size, c->superfeature);
            else if (destor.super_feature_method == SISD_FIXED_GEAR_AVX)
                sisd_gear_fixed_avx(c->data, c->size, c->superfeature);
            // else if (destor.super_feature_method == SISD_FIXED_GEAR_AVX16)
            //     sisd_gear_fixed_avx16(c->data, c->size, c->superfeature);

            TIMER_END(1, jcr.super_feature_time);
            // free_chunk(c);
        }
        // else
        //     free_chunk(c);

        sync_queue_push(resemblance_queue, c);
    }
    sync_queue_term(resemblance_queue);
    return NULL;
}

void *super_feature_thread(void *arg)
{
    struct chunk *c = (struct chunk *)arg;

    jcr.chunk_num_for_simi_dete++;
    jcr.total_size_for_similar_detection += c->size;
    if (destor.super_feature_method == SUPER_FEATURE)
        super_feature(c->data, c->size, c->superfeature);
    else if (destor.super_feature_method == SISD_FIXED_RABIN)
        Finesse(c->data, c->size, c->superfeature);
    else if (destor.super_feature_method == SISD_FIXED_GEAR)
        sisd_rabin_fixed_avx(c->data, c->size, c->superfeature);

    return NULL;
}

void send_segment_matching(struct segment *s)
{
    struct chunk *ss = new_chunk(0);
    SET_CHUNK(ss, CHUNK_SEGMENT_START);
    sync_queue_push(resemblance_queue, ss);

    GSequenceIter *end = g_sequence_get_end_iter(s->chunks);
    GSequenceIter *begin = g_sequence_get_begin_iter(s->chunks);
    while (begin != end)
    {
        struct chunk *c = g_sequence_get(begin);
        sync_queue_push(resemblance_queue, c);
        g_sequence_remove(begin);
        begin = g_sequence_get_begin_iter(s->chunks);
    }

    struct chunk *se = new_chunk(0);
    SET_CHUNK(se, CHUNK_SEGMENT_END);
    sync_queue_push(resemblance_queue, se);
    s->chunk_num = 0;
}

static void *resemblance_thread_pool(void *arg)
{
    int thread_num = destor.thread_num;
    threadpool_t *thp = threadpool_create(thread_num, 2000);
    while (1)
    {
        struct chunk *c = sync_queue_pop(dedup_queue);

        if (c == NULL)
        {
            break;
        }

        struct segment *s = new_segment();
        int unique_num = 0;
        assert(CHECK_CHUNK(c, CHUNK_SEGMENT_START));
        free_chunk(c);

        c = sync_queue_pop(dedup_queue);
        while (!(CHECK_CHUNK(c, CHUNK_SEGMENT_END)))
        {
            g_sequence_append(s->chunks, c);
            if (!CHECK_CHUNK(c, CHUNK_FILE_START) && !CHECK_CHUNK(c, CHUNK_FILE_END))
                s->chunk_num++;
            if (c->type == CHUNK_TYPE_UNIQUE_ALL)
                unique_num++;
            c = sync_queue_pop(dedup_queue);
        }

        free_chunk(c);

        int task_num = 0;
        if (unique_num > 0)
        {
            TIMER_DECLARE(1);
            TIMER_BEGIN(1);
            GSequenceIter *iter = g_sequence_get_begin_iter(s->chunks);
            GSequenceIter *end = g_sequence_get_end_iter(s->chunks);
            for (; iter != end; iter = g_sequence_iter_next(iter))
            {
                struct chunk *c = g_sequence_get(iter);
                if (CHECK_CHUNK(c, CHUNK_FILE_START) ||
                    CHECK_CHUNK(c, CHUNK_FILE_END) ||
                    CHECK_CHUNK(c, CHUNK_SEGMENT_START) ||
                    CHECK_CHUNK(c, CHUNK_SEGMENT_END) ||
                    CHECK_CHUNK(c, CHUNK_DUPLICATE) || c->size <= destor.chunk_min_size)
                {
                    continue;
                }
                task_num++;
                threadpool_add(thp, super_feature_thread, (void *)c);
            }
            wait_threadpool(thp, task_num);
            TIMER_END(1, jcr.super_feature_time);
        }
        send_segment_matching(s);
        free_segment(s);
        s = NULL;
    }
    threadpool_destroy(thp);
    sync_queue_term(resemblance_queue);

    return NULL;
}

void start_resemblance_phase()
{
    resemblance_queue = sync_queue_new(1000);
    pthread_create(&resemblance_t, NULL, resemblance_thread, NULL);
}

void stop_resemblance_phase()
{
    pthread_join(resemblance_t, NULL);
}

static void *feature_extract_thread(void *arg)
{
    while (1)
    {
        struct chunk *c = sync_queue_pop(chunk_queue);
        if (c == NULL)
            /* backup job finish */
            break;

        if (CHECK_CHUNK(c, CHUNK_FILE_START) ||
            CHECK_CHUNK(c, CHUNK_FILE_END) ||
            CHECK_CHUNK(c, CHUNK_SEGMENT_START) ||
            CHECK_CHUNK(c, CHUNK_SEGMENT_END) ||
            CHECK_CHUNK(c, CHUNK_DUPLICATE))
        {
            free_chunk(c);
            continue;
        }
        jcr.data_size += c->size;
        if (c->size > destor.chunk_min_size)
        {
            jcr.total_size_for_similar_detection += c->size;
            TIMER_DECLARE(1);
            TIMER_BEGIN(1);
            if (destor.super_feature_method == RABIN_ONLY)
                rabin_only(c->data, c->size);
            else if (destor.super_feature_method == RABIN_ONLY_8)
                rabin_only_avx8(c->data, c->size);
            else if (destor.super_feature_method == RABIN_ONLY_16)
                rabin_only_avx16(c->data, c->size);
            TIMER_END(1, jcr.super_feature_time);
            free_chunk(c);
        }
        else
            free_chunk(c);
    }
    return NULL;
}

void start_feature_extract_phase()
{
    pthread_create(&resemblance_t, NULL, feature_extract_thread, NULL);
}

void stop_feature_extract_phase()
{
    pthread_join(resemblance_t, NULL);
}
